package com.lefu.risk.storm.topology;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;
import org.apache.storm.utils.Utils;

import com.lefu.risk.storm.bolt.ReportBolt;
import com.lefu.risk.storm.bolt.WordCountBolt;
import com.lefu.risk.storm.bolt.WordSplitBolt;
import com.lefu.risk.storm.content.Content;
import com.lefu.risk.storm.spout.WordCountSpout;

public class WordCountTopology {
	
	public static void main(String[] args) throws Exception {
		TopologyBuilder builder = new TopologyBuilder();
		
		WordCountSpout wordCountSpout = new WordCountSpout();
		WordSplitBolt wordSplitBolt = new WordSplitBolt();
		WordCountBolt wordCountBolt = new WordCountBolt();
		ReportBolt reportBolt = new ReportBolt();
		
		
		builder.setSpout(Content.SENTENCE_SPOUT_ID,wordCountSpout);
		builder.setBolt(Content.SPLIT_BOLT_ID,wordSplitBolt).shuffleGrouping(Content.SENTENCE_SPOUT_ID);
		builder.setBolt(Content.COUNT_BOLT_ID,wordCountBolt).fieldsGrouping(Content.SPLIT_BOLT_ID,new Fields("word"));
		
		builder.setBolt(Content.REPORT_BOLT_ID, reportBolt).globalGrouping(Content.COUNT_BOLT_ID);
		
		StormTopology topology = builder.createTopology();
		
		Config config = new Config();
		
		LocalCluster cluster = new LocalCluster();
		cluster.submitTopology(Content.WORD_COUNT_TOPOLOGY_NAME, config, topology);
		
		Thread.sleep(15000);
		Utils.sleep(15000);
		
		cluster.killTopology(Content.WORD_COUNT_TOPOLOGY_NAME);
		cluster.shutdown();
		
//		StormSubmitter.submitTopology(Content.WORD_COUNT_TOPOLOGY_NAME, config, topology);
		
	}
}
